package com.archgeek.bigdata;

import com.alibaba.fastjson.JSONObject;
import lombok.Data;
import lombok.experimental.Accessors;
import org.apache.curator.framework.CuratorFramework;
import org.joda.time.DateTime;

import java.sql.*;
import java.util.*;

/**
 * java -cp code-hive-1.0.0-SNAPSHOT.jar com.archgeek.bigdata.hive
 *
 * @author pizhihui
 * @date 2022-02-25 14:16
 */
public class HiveTool {


    private static String driverName = "org.apache.hive.jdbc.HiveDriver";   // 此Class 位于 hive-jdbc的jar包下
    private static String DEFAULT_URL = "jdbc:hive2://10.10.29.41:11028/";    //填写hive的IP，之前在配置文件中配置的IP
    // private static Connection conn = getConnnection();
    // private static Connection conn = null;
    private static PreparedStatement ps;
    private static ResultSet rs;

    private static String URL_FORMT = "jdbc:hive2://%s/";    //填写hive的IP，之前在配置文件中配置的IP


    public static Connection getConnnection(String host) {
        Connection conn = null;
        try {
            Class.forName(driverName);
            Properties properties = new Properties();
            properties.put("user", "biadmin");
            // properties.put("socksProxy", "127.0.0.1:1081");
            conn = DriverManager.getConnection(host, properties);        //只是连接hive, 用户名可不传
        } catch (ClassNotFoundException e) {
            e.printStackTrace();
            System.exit(1);
        } catch (SQLException e) {
            e.printStackTrace();
        }
        return conn;
    }

    public static PreparedStatement prepare(Connection conn, String sql) {
        PreparedStatement ps = null;
        try {
            ps = conn.prepareStatement(sql);
        } catch (SQLException e) {
            e.printStackTrace();
        }
        return ps;
    }

    public static void testHeartBeat(Connection conn) {
        String sql = "show databases---------heartbeat -----";
        try {
            ps = prepare(conn, sql);
            rs = ps.executeQuery();
            int columns = rs.getMetaData().getColumnCount();
            while (rs.next()) {
                for (int i = 1; i <= columns; i++) {
                    System.out.print(rs.getString(i));
                    System.out.print("\t\t");
                }
                System.out.println();
            }
        } catch (SQLException e) {
            e.printStackTrace();
        }
    }


    public static void getAll(String tablename) {
        Connection conn = getConnnection(DEFAULT_URL);
        String sql = "select * from " + tablename + " limit 10 ";
        System.out.println(sql);
        try {
            ps = prepare(conn, sql);
            rs = ps.executeQuery();
            int columns = rs.getMetaData().getColumnCount();
            while (rs.next()) {
                for (int i = 1; i <= columns; i++) {
                    System.out.print(rs.getString(i));
                    System.out.print("\t\t");
                }
                System.out.println();
            }
        } catch (SQLException e) {
            e.printStackTrace();
        }

    }

    public static void getSql(String url, String sql) {
        Connection conn = getConnnection(url);
        System.out.println(DateTime.now().toLocalDateTime() + "--执行SQL....");
        // getSchemas(conn);
        try {
            ps = prepare(conn, sql);
            rs = ps.executeQuery();
            int columns = rs.getMetaData().getColumnCount();
            while (rs.next()) {
                for (int i = 1; i <= columns; i++) {
                    System.out.print(rs.getString(i));
                    System.out.print("\t\t");
                }
                System.out.println();
            }
        } catch (SQLException e) {
            e.printStackTrace();
        }
    }

    public static void getSchemas(Connection conn) {
        Set<String> res = new HashSet<>();
        ResultSet schemas = null;
        DatabaseMetaData metaData;
        try {
            metaData = conn.getMetaData();
            schemas = metaData.getSchemas();
            while (schemas.next()) {
                String schemaName = schemas.getString("TABLE_SCHEM");
                if (schemaName == null) {
                    schemaName = "";
                }
                res.add(schemaName);
            }
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            try {
                if (null != schemas) {
                    schemas.close();
                }
            } catch (SQLException e) {
                e.printStackTrace();
            }
        }
        System.out.println("schemas: " + Arrays.toString(res.toArray()));
    }

    public static List<EngineHost> getAllTs() {
        String zkAddr = "10.10.16.18:2181,10.10.16.19:2181,10.10.16.20:2181,10.10.54.12:2181,10.10.16.165:2181";
        CuratorFramework client = CuratorClient.getClient(zkAddr);

        String coreGroupPath = "/kyuubi_prod_1.5.0-SNAPSHOT_LX_GROUP_SPARK_SQL/CORE_GROUP";
        String otherGroupPath = "/kyuubi_prod_1.5.0-SNAPSHOT_LX_GROUP_SPARK_SQL/OTHER_GROUP";

        List<EngineHost> engineHostList = new ArrayList<>();

        List<EngineHost> coreList = getGroupEngineHost(client, coreGroupPath);
        List<EngineHost> otherList = getGroupEngineHost(client, otherGroupPath);

        engineHostList.addAll(coreList);
        engineHostList.addAll(otherList);

        System.out.println("最后的结果是: " + JSONObject.toJSONString(engineHostList));
        return engineHostList;
    }

    private static List<EngineHost> getGroupEngineHost(CuratorFramework client, String groupPath) {
        List<EngineHost> engineHostList = new ArrayList<>();

        try {
            List<String> enginePoolList = client.getChildren().forPath(groupPath);


            for (String enginePoolName : enginePoolList) {
                List<String> engineInfoList = client.getChildren().forPath(groupPath + "/" + enginePoolName);
                if (null != engineInfoList && !engineInfoList.isEmpty()) {
                    String engineInfo = engineInfoList.get(0);

                    String serviceHost = "";
                    String refId = "";
                    for (String s : engineInfo.split(";")) {
                        if (s.contains("serviceUri=")) {
                            serviceHost = s.split("=")[1];
                        }
                        if (s.contains("refId=")) {
                            refId = s.split("=")[1];
                        }
                    }
                    if (null != serviceHost && serviceHost.length() > 0) {
                        engineHostList.add(new EngineHost()
                                .setServiceHost(serviceHost)
                                .setRefId(refId));
                    }
                }
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
        return engineHostList;
    }

    @Data
    @Accessors(chain = true)
    static class EngineHost {
        private String serviceHost;
        private String refId;
    }


    public static void main(String[] args) {
//        String tablename = "lx_dim.dim_fin_business_sale_type_info_df";
//        getAll(tablename);
//         testHeartBeat();
        List<EngineHost> allTs = getAllTs();


        Map<String, Connection> idConnMap = new HashMap<>();
        for (EngineHost engineHost : allTs) {
            String serviceHost = engineHost.getServiceHost();
            String refId = engineHost.getRefId();
            idConnMap.put(refId, getConnnection(String.format(URL_FORMT, serviceHost)));
        }
        idConnMap.forEach((refId, conn) -> {
            try {
                testHeartBeat(conn);
                System.out.println("成功............. refId=" + refId);
            } catch (Exception e) {
                e.printStackTrace();
                System.out.println("!!!!!!!!! 出错了 !!!!!!!!!!!! refId=" + refId);
            }
        });


    }

}
